package com.xuecheng.messagesdk.service;

import com.xuecheng.messagesdk.model.po.MqMessage;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * 消息处理抽象类
 */
@Slf4j
@Data
//@Component
public abstract class MessageProcessAbstract {
    //注意，如果这里子类想直接访问mqMessageService，那么这里必须用protected或public
    //如果用private或直接default，那么子类是拿不到的，只能get
    @Autowired
    protected MqMessageService mqMessageService;

    @Resource
    Executor messageTaskExecutor;

    /**
     * 执行任务
     * @param mqMessage
     * @return
     */
    public abstract boolean execute(MqMessage mqMessage);

    /**
     * 扫描消息表多线程执行任务
     * @param shardIndex
     * @param shardTotal
     * @param messageType
     * @param count
     * @param timeout
     */
    public void process(int shardIndex,int shardTotal,String messageType,int count,long timeout){
        try{
            List<MqMessage> messageList = mqMessageService.getMessageList(shardIndex, shardTotal, messageType, count);
            int size = messageList.size();
            log.info("处理消息任务开始，本次取出了{}条消息",size);
            CountDownLatch countDownLatch = new CountDownLatch(size);
            for(int i=0;i<size;i++){
                MqMessage message = messageList.get(i);
                messageTaskExecutor.execute(() -> {
                    log.info("开始任务:{}",message);
                    try{
                        boolean result = execute(message);
                        if(result) {
                            log.info("任务执行成功:{}", message);
                            //更新任务状态
                            int completed = mqMessageService.completed(message.getId());
                            if (completed > 0) {
                                log.info("任务状态更新成功:{}", message);
                            } else {
                                log.info("任务状态更新失败:{}", message);
                            }
                        }
                    }catch(Exception e){
                        e.printStackTrace();
                        log.error("任务执行出现异常:{}",message);
                    }finally{
                        countDownLatch.countDown();
                    }
                    log.info("任务结束:{}",message);
                });
            }
            countDownLatch.await(timeout, TimeUnit.SECONDS);
            log.info("本次的{}条消息任务执行结束",size);
        }catch(Exception e){
            e.printStackTrace();
        }
    }
}
